﻿using RockBus.Configuration.Interfaces;
using RockBus.DataContracts;
using RockBus.Exceptions;
using RockBus.Logging.Interfaces;
using RockBus.MessageLogging.Interfaces;
using RockBus.Messages.Interfaces;
using RockBus.Routing.Interfaces;
using RockBus.Utilities;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Threading.Tasks;

namespace RockBus.Routing
{
    public class RoutingManager
    {
        public RoutingManager(
            ILogger logger,
            IMessageLogger messageLogger,
            IMessageFactory messageFactory,
            IQueueManager queueManager,
            ICorrelationManager correlationManager,
            IMessageProcessor[] messageProcessors
            )
        {
            this.Logger = logger;
            this.MessageLogger = messageLogger;
            this.MessageFactory = messageFactory;
            this.QueueManager = queueManager;
            this.QueueManager.Register(this.DequeueMessage, this.DequeueException, this.DequeueRetry);
            this.CorrelationManager = correlationManager;

            this.ProcessorDictionary = new Dictionary<MessageProcessorEnum, IMessageProcessor>(messageProcessors.Length);
            foreach (var mp in messageProcessors)
            {
                this.ProcessorDictionary.Add(mp.ProcessingStep, mp);
            }
        }

        private ILogger Logger { get; set; }

        private IMessageLogger MessageLogger { get; set; }

        private IMessageFactory MessageFactory { get; set; }

        private IQueueManager QueueManager { get; set; }

        private Dictionary<MessageProcessorEnum, IMessageProcessor> ProcessorDictionary { get; set; }

        private ICorrelationManager CorrelationManager { get; set; }

        private void DequeueMessage(IMessageContext messageContext)
        {
            try
            {
                this.ProcessMessageInternal(messageContext);
            }
            catch (Exception exc)
            {
                messageContext.Exception = exc;
                this.QueueManager.ProcessException(messageContext);
            }
        }

        private void DequeueException(IMessageContext messageContext)
        {
            this.ProcessException(messageContext);
        }

        private void ProcessMessageInternal(IMessageContext messageContext)
        {
            this.Logger.LogInfo(MethodBase.GetCurrentMethod(), "{0}; Processing MessageContext {1}",
                messageContext.ProcessingData.MessageDirection,
                messageContext.ToString());

            this.MessageLogger.LogMessage(messageContext);

            if (messageContext.Exception != null)
            {
                this.QueueManager.ProcessException(messageContext);
                return;
            }

            switch (messageContext.ProcessingData.MessageDirection)
            {
                case MessageDirectionEnum.Received:
                    this.ProcessorDictionary[MessageProcessorEnum.EvaluationManager].ProcessMessage(messageContext);
                    break;

                case MessageDirectionEnum.InboundRequest:
                    // Open any inbound correlations?
                    this.CorrelationManager.TryStartInboundCorrelation(messageContext);
                    if (messageContext.RoutingHeaders.IsNullOrEmpty())
                    {
                        break;
                    }

                    if (messageContext.IsPersistent())
                    {
                        // Messages with persistent subscriptions go to the PersistenceSendPort
                        this.ProcessorDictionary[MessageProcessorEnum.PersistenceSendPort].ProcessMessage(messageContext);
                    }
                    else
                    {
                        // Messages with transient subscriptions are treated as OutboundRequests
                        messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.OutboundRequest;
                        this.ProcessMessageInternal(messageContext);
                    }
                    break;

                case MessageDirectionEnum.OutboundRequest:
                    // Open any outbound correlations?
                    this.CorrelationManager.TryStartOutboundCorrelation(messageContext);

                    // Outbound Requests go to the DeliverySendPort
                    this.ProcessorDictionary[MessageProcessorEnum.DeliverySendPort].ProcessMessage(messageContext);
                    break;

                case MessageDirectionEnum.InboundResponse:
                    // Process response independently as new request
                    var newRequestContext = this.MessageFactory.CreateRequestMessageContextFromResponse(messageContext);
                    this.QueueManager.ProcessMessage(newRequestContext);

                    // Evaluate any outbound correlations
                    IMessageContext outboundCorrelatedMessageContext = null;
                    if (this.CorrelationManager.TrySatisfyOutboundCorrelation(messageContext, out outboundCorrelatedMessageContext))
                    {
                        // The message was used to satisfy a correlation with,
                        //  so this message is finished.
                        messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.End;
                        this.ProcessMessageInternal(messageContext);

                        // If an outbound correlated MessageContext was found, we continue with that
                        messageContext = outboundCorrelatedMessageContext;
                    }

                    if (messageContext.IsPersistent())
                    {
                        this.ProcessorDictionary[MessageProcessorEnum.PersistenceSendPort].ProcessMessage(messageContext);
                    }
                    else
                    {
                        messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.OutboundResponse;
                        this.ProcessMessageInternal(messageContext);
                    }
                    break;

                case MessageDirectionEnum.OutboundResponse:
                    // Evaluate any inbound correlations
                    IMessageContext inboundCorrelatedMessageContext = null;
                    if (this.CorrelationManager.TrySatisfyInboundCorrelation(messageContext, out inboundCorrelatedMessageContext))
                    {
                        // The message was used to satisfy a correlation with,
                        //  so this message is finished.
                        messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.End;
                        this.ProcessMessageInternal(messageContext);

                        // If an inbound correlated MessageContext was found, we continue with that
                        messageContext = inboundCorrelatedMessageContext;
                    }

                    // Outbound responses go back to the ReceivePort
                    this.ProcessorDictionary[MessageProcessorEnum.ReceivePort].ProcessMessage(messageContext);
                    break;

                case MessageDirectionEnum.End:
                    // Ended messages are completed
                    this.CompleteMessage(messageContext);
                    break;

                default:
                    throw new InvalidOperationException("Impossible MessageDirectionEnum!");
            }
        }

        private void CompleteMessage(IMessageContext messageContext)
        {
            this.Logger.LogInfo(MethodInfo.GetCurrentMethod(), "COMPLETING {0}", messageContext.ToString());

            // Set to End (if not already)
            messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.End;

            // Persistent messageContext should be offered to PersistenceSendPort before completion
            if (messageContext.IsPersistent())
            {
                this.ProcessorDictionary[MessageProcessorEnum.PersistenceSendPort].ProcessMessage(messageContext);
            }

            if (!messageContext.Complete())
            {
                if (messageContext.Exception == null)
                {
                    messageContext.Exception = new Exception("Unable to complete the message");
                }
                this.ProcessException(messageContext);
            }
        }

        private void ProcessException(IMessageContext messageContext)
        {
            this.Logger.LogError(MethodInfo.GetCurrentMethod(), "EXCEPTION for {0}", messageContext);

            // If the messageContext has no RoutingHeaders, it means it's an incoming message that was aborted
            //  (for instance due to a timeout of the MessageContext)
            //  Such a message should be aborted here
            if (messageContext.RoutingHeaders.IsNullOrEmpty())
            {
                messageContext.Abort();
                return;
            }

            if (messageContext.IsPersistent())
            {
                // Persistent messageContext should Abort and let the persistence store handle the retries
                //  PersistenceSendPort should also store any errors
                //  If the PersistenceSendPort does not know how to perform retries (i.e. the InMemoryPersistenceSendPort),
                //   then it should bounce the message to the RetryQueue by calling QueueManager.ProcessRetry(messageContext).
                this.ProcessorDictionary[MessageProcessorEnum.PersistenceSendPort].ProcessMessage(messageContext);
            }
            else
            {
                // Non-persistent should start/follow an in-memory retry mechanism
                this.QueueManager.ProcessRetry(messageContext);
            }
        }

        public void DequeueRetry(IMessageContext messageContext)
        {
            this.ProcessRetry(messageContext);
        }

        private async Task ProcessRetry(IMessageContext messageContext)
        {
            try
            {
                if (messageContext.RoutingHeaders.IsNullOrEmpty() ||
                    (messageContext.Exception is RoutingException))
                {
                    this.Logger.LogInfo(MethodBase.GetCurrentMethod(), "Ignoring retry. RoutingHeaders.IsNullOrEmpty(): {0}; Exception is RoutingException: {1}",
                        messageContext.RoutingHeaders.IsNullOrEmpty(),
                        messageContext.Exception is RoutingException);

                    // If there are no subscriptions, and/or if it concerns a RoutingException,
                    //  there's no point in retrying, so we abort immediately
                    messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.End;
                    this.ProcessorDictionary[MessageProcessorEnum.ErrorManager].ProcessMessage(messageContext);
                    return;
                }

                // If we got here, the DeliveryAttemptCount is 1 at minimum
                int deliveryAttemptCount = 1;

                // Get the current DeliveryAttemptCount if it exists
                object deliveryAttemptCountObj = null;
                if (messageContext.GetProperty(RoutingHeader.DeliveryAttemptCountKey, out deliveryAttemptCountObj))
                {
                    deliveryAttemptCount = (int)deliveryAttemptCountObj;
                    ++deliveryAttemptCount;
                }

                messageContext.SetProperty(RoutingHeader.DeliveryAttemptCountKey, deliveryAttemptCount);

                // Get the MaxDeliveryAttemptCount and Interval
                var rh = messageContext.RoutingHeaders.First();
                if (deliveryAttemptCount < rh.SubscriptionDetail.MaxDeliveryAttemptCount)
                {
                    long timeoutMs = rh.SubscriptionDetail.DeliveryAttemptIntervalSecs.Value * 1000;
                    if (timeoutMs > int.MaxValue)
                    {
                        timeoutMs = int.MaxValue;
                    }

                    this.Logger.LogInfo(MethodBase.GetCurrentMethod(), "Going into retry wait interval {0} secs for {1} with attemptCount {2}",
                        rh.SubscriptionDetail.DeliveryAttemptIntervalSecs.Value,
                        messageContext,
                        deliveryAttemptCount);
                    await Task.Delay((int)timeoutMs);

                    // Reset exception
                    this.Logger.LogInfo(MethodBase.GetCurrentMethod(), "Retrying message. DeliveryAttemptCount: {0} for {1}",
                        deliveryAttemptCount,
                        messageContext);
                    messageContext.Exception = null;
                    this.QueueManager.ProcessMessage(messageContext);
                }
                else
                {
                    // End of the line for this messageContext
                    this.Logger.LogError(MethodInfo.GetCurrentMethod(), "EXCEPTION: MaxDeliveryAttemptCount {0} exhausted, message discarded for {1}",
                        rh.SubscriptionDetail.MaxDeliveryAttemptCount,
                        messageContext);

                    // Offer the MessageContext to the ErrorManager for one last time,
                    //  so it can log the error if possible
                    messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.End;
                    this.ProcessorDictionary[MessageProcessorEnum.ErrorManager].ProcessMessage(messageContext);
                }
            }
            catch (Exception exc)
            {
                this.Logger.Log(MethodBase.GetCurrentMethod(), exc);

                messageContext.Exception = exc;
                this.QueueManager.ProcessException(messageContext);
            }
        }
    }
}